package cn.doitedu.data_mock;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.codec.digest.Md5Crypt;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.UUID;

public class VideoDataMock {

    public static void main(String[] args) throws InterruptedException {

        for(int i=0;i<=1000;i++) {
            new Thread(new MyRunner((i+1)*1000,(i+2)*1000)).start();
            Thread.sleep(50);
        }

    }



    public static class MyRunner implements Runnable{

        int id1;
        int id2;
        String topicName = "dwd-user-action";
        KafkaProducer<String, String> producer;

        JSONObject jsonObject;
        JSONObject props;
        public MyRunner(int id1,int id2){
            this.id1 = id1;
            this.id2 = id2;

            Properties properties = new Properties();
            properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"doitedu:9092");
            properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.setProperty(ProducerConfig.ACKS_CONFIG,"0");

            producer = new KafkaProducer<String, String>(properties);


            jsonObject = new JSONObject();
            props = new JSONObject();
            props.put("url","");
            props.put("video_id",1L);
            props.put("play_id","");

            jsonObject.put("properties",props);
            jsonObject.put("user_id",1L);
            jsonObject.put("session_id",1L);
            jsonObject.put("event_id","");
            jsonObject.put("action_time",1L);
            jsonObject.put("release_channel","");
            jsonObject.put("device_type","");
            jsonObject.put("province","江苏省");
            jsonObject.put("city","南京市");
            jsonObject.put("region","秦淮区");
        }

        @Override
        public void run() {

            for(int i=0;i<1000;i++){
                String sessionId = RandomStringUtils.randomAlphabetic(14).toLowerCase();
                jsonObject.put("session_id",sessionId);

                int user_id = RandomUtils.nextInt(id1,id2);
                int videoId = RandomUtils.nextInt(1,100);

                String playId = RandomStringUtils.randomAlphabetic(10).toUpperCase();

                long action_time = System.currentTimeMillis();
                jsonObject.put("user_id",user_id);
                jsonObject.put("action_time",action_time);
                jsonObject.put("event_id","video_play");
                props.put("video_id",videoId);
                props.put("play_id",playId);

                //System.out.println(data_play);
                producer.send(new ProducerRecord<>(topicName, jsonObject.toJSONString()));

                for(int j=0;j<RandomUtils.nextInt(1,10);j++){
                    action_time = System.currentTimeMillis();
                    //System.out.println(data_hb);
                    jsonObject.put("action_time",action_time);
                    jsonObject.put("event_id","video_hb");

                    producer.send(new ProducerRecord<>(topicName, jsonObject.toJSONString()));
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }

                action_time = System.currentTimeMillis();
                jsonObject.put("action_time",action_time);
                jsonObject.put("event_id","video_pause");

                producer.send(new ProducerRecord<>(topicName, jsonObject.toJSONString()));
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                action_time = System.currentTimeMillis();
                jsonObject.put("action_time",action_time);
                jsonObject.put("event_id","video_resume");

                producer.send(new ProducerRecord<>(topicName, jsonObject.toJSONString()));
                try {
                    Thread.sleep(RandomUtils.nextInt(100,200));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                for(int j=0;j<RandomUtils.nextInt(1,10);j++){
                    action_time = System.currentTimeMillis();
                    jsonObject.put("action_time",action_time);
                    jsonObject.put("event_id","video_hb");

                    producer.send(new ProducerRecord<>(topicName, jsonObject.toJSONString()));
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }


                action_time = System.currentTimeMillis();
                jsonObject.put("action_time",action_time);
                jsonObject.put("event_id","video_stop");

                producer.send(new ProducerRecord<>(topicName, jsonObject.toJSONString()));
                producer.flush();
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}
